package com.zhz.elasticjob.job;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
import com.dangdang.ddframe.job.lite.api.strategy.impl.AverageAllocationJobShardingStrategy;
import com.zhz.elasticjob.annotation.ElasticDataflowJob;
import com.zhz.elasticjob.listener.ZhzListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @program: schedule
 * @description: 简单的任务
 * @author: zhanghz001
 * @create: 2021-08-05 09:53
 **/
@Slf4j
@Component
@ElasticDataflowJob(name = "dataflowJob",
        cron = "0/10 * * * * ?",
        shardingTotalCount = 2,
        overwrite = true,
        streamingProcess = true,
        listeners = {ZhzListener.class},
        jobShardingStrategy = AverageAllocationJobShardingStrategy.class,
        failover = true,
        monitorExecution = true)
public class MyDataflowJob implements DataflowJob<Integer> {
    
    @Override
    public List<Integer> fetchData(ShardingContext shardingContext) {
        //负责选取需要执行的list
        int shardingItem = shardingContext.getShardingItem();
        //
        List<Integer> integers = new ArrayList<>();
        integers.add(shardingItem);
        if (shardingItem == 0) {
            log.info("dataflowJob 任务执行了,分片项: {} , 总分片 : {} ",
                    shardingContext.getShardingItem(),
                    shardingContext.getShardingTotalCount());
            log.info("看看streamingProcess=true, 但是返回null的时候,是否会一直调用");
            return null;
        }
        return integers;
    }
    
    @Override
    public void processData(ShardingContext shardingContext, List<Integer> list) {
        //执行
        log.info("dataflowJob 任务执行了,分片项: {} , 总分片 : {}, 执行个的list: {}",
                shardingContext.getShardingItem(),
                shardingContext.getShardingTotalCount(), list);
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
